package com.bwie.service.impl;

import cn.hutool.core.util.RandomUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.bwie.config.KafkaConfig;
import com.bwie.mapper.ReportMapper;
import com.bwie.pojo.Report;
import com.bwie.pojo.ReportNation;
import com.bwie.service.ReportService;
import com.bwie.utils.MobileMessageUtil;
import com.bwie.utils.R;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @Author GaoYiJie
 * @Description
 * @Create 2025/4/17 11:51
 * @Version 1.0
 */
@Service
public class ReportServiceImpl implements ReportService {

    @Resource
    ReportMapper reportMapper;

    @Resource
    StringRedisTemplate stringRedisTemplate;

    @Resource
    KafkaTemplate kafkaTemplate;

    @Override
    public R report(ReportNation reportNation) {
        //先去缓存当中去进行判断是否存在,根据身份证号进行判断
        String reportNationsKey = "report_list";
        String reportNationListRedis = stringRedisTemplate.opsForValue().get(reportNationsKey);

        List<ReportNation> reportNations = JSON.parseArray(reportNationListRedis, ReportNation.class);
        for (ReportNation reportNationDB : reportNations) {
            //身份证相同,比赛相同,则不能落库
            if (reportNation.getReportCard().equals(reportNationDB.getReportCard()) && reportNation.getGameType().equals(reportNationDB.getGameType())) {
                return R.ERROR(500, "上报数据已存在");
            }
        }


        //redis分布式锁解决，数据重复提交的问题
        //获取锁
        String lockKey = "lock_report";
        Boolean hasLock = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "1", 1, TimeUnit.MINUTES);
        try {
            if (!hasLock) {
                //获取锁失败,休眠一段时间重试
                Thread.sleep(100);
                return report(reportNation);
            }


            //获取锁成功
            //上报
            reportMapper.insert(reportNation);

            //重建缓存
            this.initReportToRedis();

            //MQ发送短信通知青少年
            kafkaTemplate.send(KafkaConfig.TOPIC_REPORT,reportNation.getReportPhone());
        } catch (Exception e) {
            return R.ERROR(500, "上报失败");
        } finally {
            //释放锁
            stringRedisTemplate.delete(lockKey);
        }
        return R.OK();
    }

    //缓存重建
    public void initReportToRedis() {
        List<ReportNation> reportNations = reportMapper.selectList(null);

        //生成随机数 :  预防缓存雪崩
        int randomInt = RandomUtil.randomInt(0, 10);

        String reportNationsKey = "report_list";
        stringRedisTemplate.opsForValue().set(reportNationsKey, JSON.toJSONString(reportNations), 30 + randomInt, TimeUnit.MINUTES);
    }

    @Override
    public R getReportNationList(Integer id) {
        //先去缓存当中获取
        String reportNationKey = "report:" + id;
        String reportNationRedis = stringRedisTemplate.opsForValue().get(reportNationKey);
        if (StrUtil.isNotBlank(reportNationRedis)) {
            //不是空串或null,有数据
            return R.OK(JSON.parseObject(reportNationRedis, ReportNation.class));
        }

        //空值,进行拦截  预防缓存穿透
        if ("".equals(reportNationRedis)) {
            return R.ERROR(500, "该上报人不存在");
        }

        //从数据库获取
        ReportNation reportNation = reportMapper.selectById(id);
        if (reportNation == null) {
            //将空值,存入缓存 5秒过期 : 预防缓存穿透  为了测试设置为1分钟
            stringRedisTemplate.opsForValue().set(reportNationKey, "", 1, TimeUnit.MINUTES);
            return R.ERROR(500, "该上报人不存在");
        }
        stringRedisTemplate.opsForValue().set(reportNationKey, JSON.toJSONString(reportNation), 5, TimeUnit.MINUTES);
        return R.OK(reportNation);
    }


    @KafkaListener(topics = KafkaConfig.TOPIC_REPORT)
    public void receiveMessage(String message, Acknowledgment acknowledgment){
        try {
            System.out.println("消息:"+ message);

            MobileMessageUtil.sendMobileMessage(message,"1234",1);
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            //消息确认
            acknowledgment.acknowledge();
        }

    }
}
